package com.xxmw.transaction.service.app.component;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.xxmw.transaction.common.utils.bourse.PoloniexUtil;
import com.xxmw.transaction.enumeration.BourseEnum;
import com.xxmw.transaction.model.app.BoursePrice;
import com.xxmw.transaction.model.app.Symbol;
import com.xxmw.transaction.service.app.SymbolService;
import com.xxmw.transaction.service.support.AbstractWebSocketAdapter;
import com.xxmw.transaction.service.support.SymbolSubWebSocket;
import lombok.extern.slf4j.Slf4j;
import org.java_websocket.client.WebSocketClient;
import org.java_websocket.drafts.Draft;
import org.java_websocket.drafts.Draft_6455;
import org.java_websocket.handshake.ServerHandshake;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.File;
import java.io.FileInputStream;
import java.math.BigDecimal;
import java.net.URI;
import java.nio.charset.StandardCharsets;
import java.util.Date;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;

/**
 * P网 websocket api
 * https://docs.poloniex.com/#account-notifications-beta
 * @author luowenliang
 * 存在问题只返回一次有效数据
 * @since 2020/3/26 16:07
 */
@Slf4j
@Component
public final class PoloniexApplication extends AbstractWebSocketAdapter implements SymbolSubWebSocket {

    private static final String WS_URL = "wss://api2.poloniex.com";

    @Autowired
    private SymbolService symbolService;

    /**
     * 执行心跳的间隔时间和在此基础商上的超时时间将由实现的子类所设定(或者以配置的方式来是定)
     * 假如：此时间有父类适配器所设定那么所有共同子类将继承，如需改变其中一个子类的时间，将会造成
     * 所有子类间隔时间更改
     */
    private long period = 3000;


    @Override
    public BoursePrice getSymbolPrice(String symbol) {
        return symbol_map.get(symbol);
    }

    @Override
    public void cleanSymbolPrice(String symbol) {
    	symbol_map.remove(symbol);
    }

    /**
     * { "command": "subscribe", "channel": "USDT_BTC" }
     * @param symbol
     */
    @Override
    public void subscribe(Symbol symbol) {
        JSONObject object = new JSONObject();
        object.put("command","subscribe");
        object.put("channel", BourseEnum.getFormatByBourse(BourseEnum.POLONIEX.getValue()).symbolFormat(symbol.getBaseCurrency(), symbol.getQuoteCurrency()));
        this.client.send(object.toJSONString());
    }

    @Override
    public void unSubscribe(Symbol symbol) {
        JSONObject object = new JSONObject();
        object.put("command","unsubscribe");
        object.put("channel", BourseEnum.getFormatByBourse(BourseEnum.POLONIEX.getValue()).symbolFormat(symbol.getBaseCurrency(), symbol.getQuoteCurrency()));
        this.client.send(object.toJSONString());
    }

    @Override
    public void subscribeList(Set<Symbol> subSet) {
        for (Symbol symbol: subSet) {
            subscribe(symbol);
        }
    }

    @Override
    public void destroy() {
        this.client.close();
        scheduled.shutdown();
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        super.afterPropertiesSet(new PoloniexApplication.PoloniexWebSocketClient(new URI(WS_URL), new Draft_6455(), null, 5000));
        scheduled.scheduleWithFixedDelay(() -> heartBeat(), period, period, TimeUnit.MILLISECONDS);
    }


    private final class PoloniexWebSocketClient extends WebSocketClient {

        public PoloniexWebSocketClient(URI serverUri, Draft protocolDraft, Map<String, String> httpHeaders, int connectTimeout) {
            super(serverUri, protocolDraft, httpHeaders, connectTimeout);
        }

        @Override
        public void onOpen(ServerHandshake serverHandshake) {
            log.info("P网开启连接");
            Set<Symbol> set = symbolService.listByBourse(BourseEnum.POLONIEX.getValue());
            subscribeList(set);
        }

        @Override
        public void onMessage(String s) {
            // log.info("P网接收到数据:{}",s);
            //只处理更新推送数据
            if(null != s && s.contains("o")){
                JSONArray res = JSONObject.parseArray(s);
                int symbolId = res.getInteger(0);
                Map<String, Object> symbolMap = PoloniexUtil.getSymbol(symbolId);
                if (null != symbolMap.get("outSymbol")) {
                    BigDecimal ask =  null;
                    BigDecimal bid =  null;

                    JSONArray updateTicker = res.getJSONArray(2);
                    for (int i = 0; i < updateTicker.size(); i++) {
                        JSONArray ticker = updateTicker.getJSONArray(i);
                        if ("o".equals(ticker.getString(0))) {
                            //1 for bid 0 for ask 更新的数据最高买价在最后 最低卖价也在最后 遍历直接覆盖即可
                            if ("1".equals(ticker.getString(1))) {
                                bid = ticker.getBigDecimal(2);
                            }
                            if ("0".equals(ticker.getString(1))) {
                                ask = ticker.getBigDecimal(2);
                            }
                        }
                    }
                    BoursePrice oldPrice = symbol_map.get(symbolMap.get("symbol"));
                    BoursePrice price = new BoursePrice();
                    price.setData(s);
                    price.setUpdateTime(new Date());
                    if (null == oldPrice) {
                        price.setBuyPrice(bid);
                        price.setSellPrice(ask);
                    } else {
                        price.setBuyPrice(oldPrice.getBuyPrice());
                        price.setSellPrice(oldPrice.getSellPrice());
                        if (null != ask) {
                            price.setSellPrice(ask);
                        }
                        if (null != bid) {
                            price.setBuyPrice(bid);
                        }
                    }
                    // log.info("P网获取到【***{}***】的实时价格为：{}", symbolMap.get("symbol"), price.toString());
                    symbol_map.put((String)symbolMap.get("symbol"), price);
                }
            } else {
                log.error("P网websocket未处理该消息");
            }
        }

        @Override
        public void onClose(int i, String s, boolean b) {
            log.info("P网Websocket接口已关闭, 原因: {}, 状态: {}", s, i);
        }

        @Override
        public void onError(Exception e) {
            log.error("P网WebSocket接口，发生异常", e);
        }
    }

    public static void main(String[] args) throws Exception {

        File file = new File("/Users/skyruo/Downloads/st.txt");

        FileInputStream inputStream = new FileInputStream(file);
        int length = inputStream.available();
        byte bytes[] = new byte[length];
        inputStream.read(bytes);
        inputStream.close();
        String s = new String(bytes, StandardCharsets.UTF_8);

        JSONArray res = JSONObject.parseArray(s);
        BigDecimal ask =  null;
        BigDecimal bid =  null;

        //第一次全量数据推送
        JSONArray dataArray = res.getJSONArray(2).getJSONArray(0);
        JSONObject dataObject = dataArray.getJSONObject(1);
        JSONArray orderBook = dataObject.getJSONArray("orderBook");

        //这里转成JSONObject对象顺序会改变无法确定最高最低价
        String askTicker = orderBook.getString(0);
        String askStr = askTicker.substring(2, askTicker.indexOf(":") - 1);
        String bidTicker = orderBook.getString(1);
        String bidStr = askTicker.substring(2, bidTicker.indexOf(":") - 1);
        ask =  new BigDecimal(askStr);
        bid =  new BigDecimal(bidStr);
        System.out.println(ask);
        System.out.println(bid);
    }
}
